#include <stdio.h>
#include <pthread.h>
#include <malloc.h>
#include "priority_queue.h"
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include <sys/time.h>

volatile static unsigned char threadpool_task_queue_max;

typedef struct threadpool
{
    int max_task_num;
    int max_threads_num;
    int min_threads_num;
    int threads_num;
    PriorityQueue pq;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int status;
    pthread_t master;
    pthread_t thread_id[10];
} ThreadPool;

time_t get_timeval_ms()
{
    struct timeval time_now = {0};
    time_t time_s = 0;  // 秒
    time_t time_ms = 0; // 1毫秒 = 1秒/1000
    time_t time_us = 0; // 1微秒 = 1毫秒/1000

    gettimeofday(&time_now, NULL);
    time_s = time_now.tv_sec;
    time_ms = time_s * 1000 + time_now.tv_usec / 1000;
    time_us = time_now.tv_sec * 1000 * 1000 + time_now.tv_usec;
    return time_us;
}
void thread_pool_destroy(ThreadPool *threadpool)
{
    if (NULL == threadpool)
    {
        return;
    }
    printf("threadpool destroy ...pthread_mutex_lock\n");

    pthread_mutex_lock(&threadpool->mutex);
    threadpool->status = 0; // 将线程池状态设置为停止
    printf("threadpool destroy ...pthread_cond_broadcast\n");
    pthread_cond_broadcast(&threadpool->cond); // 唤醒所有等待的线程
    pthread_mutex_unlock(&threadpool->mutex);

    // for (int i = 0; i < 10; i++)
    // {
    //     // 等待所有线程执行完毕
    //     pthread_join(threadpool->thread_id[i], NULL);
    // }

    pthread_join(threadpool->master, NULL);

    printf("threadpool destroy ...\n");

    priority_queue_destroy(threadpool->pq);
    pthread_mutex_destroy(&threadpool->mutex);
    pthread_cond_destroy(&threadpool->cond);
    free(threadpool);
    return;
}

int thread_pool_add_task(ThreadPool *threadpool, void (*func)(void *), void *args, int priority)
{
    if (NULL == threadpool || NULL == func)
    {
        return -1;
    }

    Task task;
    task.func = func;
    task.args = args;
    task.priority = priority;
    task.time = get_timeval_ms();
    task.status = 0;

    pthread_mutex_lock(&threadpool->mutex);

    if (0 == priority_queue_insert(threadpool->pq, task))
    {
        printf("threadpool add task success\n");
    }
    else
    {
        pthread_mutex_unlock(&threadpool->mutex);
        goto ERR;
    }

    pthread_cond_signal(&threadpool->cond);
    pthread_mutex_unlock(&threadpool->mutex);

    return 0;
ERR:
    printf("threadpool add task fail\n");
    return -1;
}

int thread_pool_get_task_num(ThreadPool *threadpool)
{
    if (NULL == threadpool)
    {
        return -1;
    }

    return threadpool->pq->size;
}

void thread_pool_exex_task(ThreadPool *threadpool)
{
    if (NULL == threadpool)
    {
        return;
    }

    Task task;
    while (1)
    {
        pthread_mutex_lock(&threadpool->mutex);
        while (0 == threadpool->pq->size && threadpool->status)
        {
            if ((threadpool->min_threads_num > threadpool->pq->size && threadpool->threads_num > threadpool->min_threads_num))
            {
                break;
            }
            // printf("thread wait \n");
            pthread_cond_wait(&threadpool->cond, &threadpool->mutex);
        }
        if (0 == threadpool->status || (threadpool->min_threads_num > threadpool->pq->size && threadpool->threads_num > threadpool->min_threads_num))
        {
            printf("thread exit\n");
            threadpool->threads_num--;
            pthread_mutex_unlock(&threadpool->mutex);
            break;
        }
        task = priority_queue_extract(threadpool->pq);
        pthread_mutex_unlock(&threadpool->mutex);
        if (NULL == task.func)
        {
            printf("[%s:%d][\033[1;31m%s\033[0m]--! task is null !--\n", __FILE__, __LINE__, __FUNCTION__);
            continue;
        }
        task.func(task.args);
        printf("[%s:%d][\033[1;31m%s\033[0m]--! priority %d [%ld]exec time %ldus !--\n", __FILE__, __LINE__, __FUNCTION__, task.priority, task.time, task.time - get_timeval_ms());
    }
    pthread_exit(NULL);
}

void thread_pool_master(void *arg)
{
    ThreadPool *threadpool = (ThreadPool *)arg;
    while (1)
    {
        pthread_mutex_lock(&threadpool->mutex);
        if (threadpool->threads_num > threadpool->min_threads_num && threadpool->min_threads_num > threadpool->pq->size)
        {
            pthread_cond_signal(&threadpool->cond);
            printf("signal!\n");
        }

        if (threadpool->threads_num < threadpool->max_threads_num && threadpool->threads_num < threadpool->pq->size)
        {
            pthread_t thread_id;
            pthread_create(&thread_id, NULL, (void *)thread_pool_exex_task, (void *)threadpool);
            threadpool->thread_id[threadpool->threads_num++] = thread_id;
            printf("master create thread\n");
        }
        pthread_mutex_unlock(&threadpool->mutex);
        if (0 == threadpool->status)
        {
            break;
        }
        usleep(1000);
    }

    while (threadpool->threads_num)
    {
        pthread_join(threadpool->thread_id[threadpool->threads_num], NULL);
    }

    pthread_exit(NULL);
}
ThreadPool *thread_pool_init(int task_queue_num, int min_threads_num, int max_threads_num)
{
    if (task_queue_num <= 0 || min_threads_num <= 0 || max_threads_num <= 0)
    {
        return NULL;
    }

    ThreadPool *threadpool = (ThreadPool *)malloc(sizeof(ThreadPool) + sizeof(int) * max_threads_num);
    if (NULL == threadpool)
    {
        return NULL;
    }

    threadpool->max_task_num = task_queue_num;
    threadpool->max_threads_num = max_threads_num;
    threadpool->min_threads_num = min_threads_num;
    threadpool->status = 1;
    threadpool->threads_num = 0;
    threadpool->pq = priority_queue_init(task_queue_num);
    if (NULL == threadpool->pq)
    {
        free(threadpool);
        return NULL;
    }

    pthread_mutex_init(&threadpool->mutex, NULL);
    pthread_cond_init(&threadpool->cond, NULL);

    pthread_create(&threadpool->master, NULL, (void *)thread_pool_master, (void *)threadpool);

    for (int i = 0; i < min_threads_num; i++)
    {
        printf("[%s:%d][\033[1;31m%s\033[0m]--! create thread %d !--\n", __FILE__, __LINE__, __FUNCTION__, i);
        pthread_create(&threadpool->thread_id[i], NULL, (void *)thread_pool_exex_task, (void *)threadpool);
        threadpool->threads_num += 1;
    }

    return threadpool;
}

void test_task(void *args)
{
    // printf("[%s:%d][\033[1;31m%s\033[0m]--! task !--\n", __FILE__, __LINE__, __FUNCTION__);
    usleep(1000);
}

#define MACRO2
#define MACRO4

#if (defined MACRO1 || defined MACRO2 || defined MACRO3) && defined MACRO4
    #pragma message("MACRO1, MACRO2, MACRO3, MACRO4")
#endif
int main()
{
    ThreadPool *threadpool = thread_pool_init(50, 10, 10);
    if (NULL == threadpool)
    {
        return -1;
    }

    priority_queue_setdebug_leval(PRIORITYQUEUE_CLOSE);
    sleep(5);
    // thread_pool_add_task(threadpool, priority_queue_test_main, NULL, 1);
    // thread_pool_add_task(threadpool, priority_queue_test_main, NULL, 0);
    for (int i = 0; i < 50; i++)
    {

        thread_pool_add_task(threadpool, test_task, NULL, i % 5);
    }

    sleep(10);
    printf("thread_pool_destroy\n");
    sleep(10);
    thread_pool_destroy(threadpool);
    return 0;
}